package com.sh.data.engine.domain.common.service.impl;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.exec.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.io.*;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.function.Consumer;
import java.util.stream.Collectors;

@Slf4j
@Service
public class CommandExecutorServiceImpl {

    public static final Long DEFAULT_TIMEOUT = 30000L;
    private static final Logger logger = LoggerFactory.getLogger(CommandExecutorServiceImpl.class);

    private ExecutorService pool;

    @PostConstruct
    public void init() {
        ThreadFactory threadFactory =
            new ThreadFactoryBuilder().setNameFormat("command-dev-pool").build();
        pool =
            new ThreadPoolExecutor(
                2,
                5,
                0L,
                TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(),
                threadFactory,
                new ThreadPoolExecutor.AbortPolicy());
    }

    @PreDestroy
    public void destroy() {
        if (pool != null) {
            pool.shutdownNow();
        }
    }

    public ExecuteResult executeCommand(String command) {
        return executeCommand(command, null, null, DEFAULT_TIMEOUT);
    }

    public ExecuteResult executeCommand(String command, Map<String, String> envMap, File workingDir) {
        return executeCommand(command, envMap, workingDir, DEFAULT_TIMEOUT);
    }

    public ExecuteResult executeCommand(
        String command, Map<String, String> envMap, File workingDir, Long timeout) {
        boolean isWindows = System.getProperty("os.name").toLowerCase().startsWith("windows");
        String[] cmds =
            isWindows ? new String[]{"cmd.exe", "/c", command} : new String[]{"sh", "-c", command};

        String[] envp = null;
        if (envMap != null && !envMap.isEmpty()) {
            List<String> envList =
                envMap.entrySet().stream()
                    .collect(
                        Collectors.mapping(
                            entry -> String.format("%s=%s", entry.getKey(), entry.getValue()),
                            Collectors.toList()));
            envp = envList.toArray(new String[envList.size()]);
        }

        File dirToUse = (workingDir == null) ? new File(System.getProperty("user.home")) : workingDir;
        long timeOutToUse = (timeout == null || timeout <= 0) ? DEFAULT_TIMEOUT : timeout;

        Process process = null;
        try {
            process = Runtime.getRuntime().exec(cmds, envp, dirToUse);
            // 关闭outputstream，处理inputstream和errorstream
            process.getOutputStream().close();
            try (InputStream pIn = process.getInputStream();
                 InputStream pErr = process.getErrorStream();) {
                StringBuilder buffer = new StringBuilder(), bufferError = new StringBuilder();
                CountDownLatch latch = new CountDownLatch(2);

                Future<?> inFuture = pool.submit(new StreamGobbler(pIn, buffer::append, latch));
                Future<?> errorFuture = pool.submit(new StreamGobbler(pErr, bufferError::append, latch));
                boolean rst = process.waitFor(timeOutToUse, TimeUnit.MILLISECONDS);
                if (!rst) {
                    logger.error("The command [{}] timed out.", command);
                    inFuture.cancel(true);
                    errorFuture.cancel(true);
                    return new ExecuteResult(-1, bufferError.toString());
                }
                latch.await();

                int exitCode = process.exitValue();

                if (exitCode == 0 || buffer.length() > 0) {
                    logger.info("The command [{}]", command);
                    logger.info("exitCode:{} ; error:{}", exitCode, bufferError.toString());
                    return new ExecuteResult(exitCode, buffer.toString());
                } else {
                    logger.info("The command [{}]", command);
                    logger.info("exitCode:{} ; error:{}", exitCode, bufferError.toString());
                    return new ExecuteResult(exitCode, bufferError.toString());
                }
            } catch (InterruptedException e) {
                logger.error("The command [{}] did not complete due to an interrupted error.", command, e);
                return new ExecuteResult(-1, e.getMessage());
            } catch (IOException e1) {
                logger.error("The command [{}] execute failed due to io error.", command, e1);
                return new ExecuteResult(-1, e1.getMessage());
            }
        } catch (IOException eio) {
            logger.error("The command [{}] execute failed due to io error.", command, eio);
            return new ExecuteResult(-1, eio.getMessage());
        } finally {
            if (process != null) {
                process.destroy();
            }
        }
    }

    public ExecuteResult executeAddUserCommand(String cmd) {
        ExecuteResult executeResult = this.executeCommand(cmd);
        // 用户已经存在
        if (executeResult.getExitCode() == 9) {
            executeResult.setExitCode(0);
        }
        return executeResult;
    }

    public ExecuteResult executeCreateHdfsUserDirCommand(String cmd) {
        ExecuteResult executeResult = this.executeCommand(cmd);
        // 用户已经存在
        if (executeResult.getExitCode() == 9) {
            executeResult.setExitCode(0);
        }
        return executeResult;
    }

    public static class StreamGobbler implements Runnable {
        private InputStream inputStream;
        private Consumer<String> consumer;
        private CountDownLatch countDownLatch;

        public StreamGobbler(
            InputStream inputStream, Consumer<String> consumer, CountDownLatch countDownLatch) {
            this.inputStream = inputStream;
            this.consumer = consumer;
            this.countDownLatch = countDownLatch;
        }

        @Override
        public void run() {
            try {
                if (inputStream != null) {
                    try (InputStream isToUse = inputStream;
                         BufferedReader reader = new BufferedReader(new InputStreamReader(isToUse, "UTF-8"))) {
                        reader.lines().forEach(consumer);
                    } catch (IOException e) {
                        log.error(e.getMessage(), e);
                    }
                }
            } finally {
                countDownLatch.countDown();
            }
        }
    }

    public int executeShell(String shellFile, String username, LogOutputStream logOutputStream) {
        boolean isWindows = System.getProperty("os.name").toLowerCase().startsWith("windows");

        String executable = isWindows ? "cmd.exe" : "sudo";

        String[] params =
            isWindows ? new String[]{shellFile} : new String[]{"-u", username, "sh", shellFile};

        return execute(executable, logOutputStream, params);
    }

    public int execute(String executable, LogOutputStream logOutputStream, String... params) {

        ExecuteWatchdog watchdog = new ExecuteWatchdog(1000 * 60 * 60 * 24);

        DefaultExecutor exec = new DefaultExecutor();
        exec.setStreamHandler(new PumpStreamHandler(logOutputStream));
        exec.setWatchdog(watchdog);
        CommandLine commandLine = new CommandLine(executable);
        if (params != null && params.length > 0) {
            commandLine.addArguments(params, true);
        }
        try {
            int exitValue = exec.execute(commandLine);
            return exitValue;
        } catch (IOException e) {
            log.error("执行命令异常", e);
            return -1;
        }
    }

    public static class ExecuteResult {
        private int exitCode;
        private String executeOut;

        public ExecuteResult(int exitCode, String executeOut) {
            this.exitCode = exitCode;
            this.executeOut = executeOut;
        }

        public int getExitCode() {
            return exitCode;
        }

        public void setExitCode(int exitCode) {
            this.exitCode = exitCode;
        }

        public String getExecuteOut() {
            return executeOut;
        }

        public void setExecuteOut(String executeOut) {
            this.executeOut = executeOut;
        }

        public boolean isSuccess() {
            return exitCode == 0;
        }

        @Override
        public String toString() {
            return "ExecuteResult [exitCode=" + exitCode + ", executeOut=" + executeOut + "]";
        }
    }
}
